package fun.tan90.easy.binlog.worker.handler.listener;


import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.github.shyiko.mysql.binlog.event.*;
import fun.tan90.easy.binlog.worker.handler.processor.BinlogEventProcessor;
import fun.tan90.easy.binlog.worker.model.BinlogTableMap;
import lombok.extern.slf4j.Slf4j;

import java.util.Date;
import java.util.HashMap;
import java.util.Map;

/**
 * description TODO
 * company 铁人科技
 *
 * @author 陈浩杰
 * @date 2023/9/1 12:53
 */
@Slf4j
public class BinlogEventListener implements BinaryLogClient.EventListener {

    private static final Map<Long, BinlogTableMap> TABLE_MAP = new HashMap<>();

    /**
     * slave客户端编号
     */
    private final String masterCode;

    /**
     * 时间偏移量 ms
     */
    private final Integer offsetMilliseconds;

    /**
     * binlog处理器
     */
    private final BinlogEventProcessor binlogEventProcessor;

    public BinlogEventListener(String masterCode, Integer offsetMilliseconds, BinlogEventProcessor binlogEventProcessor) {
        this.masterCode = masterCode;
        this.offsetMilliseconds = offsetMilliseconds;
        this.binlogEventProcessor = binlogEventProcessor;
    }

    @Override
    public void onEvent(Event event) {
        EventHeaderV4 header = event.getHeader();
        EventType eventType = header.getEventType();
        long timestamp = header.getTimestamp();
        if (EventType.TABLE_MAP == eventType) {
            TableMapEventData data = event.getData();
            TABLE_MAP.put(data.getTableId(), BinlogTableMap.builder().database(data.getDatabase()).table(data.getTable()).build());
        } else if (EventType.isRowMutation(eventType)) {
            if (EventType.isWrite(eventType)) {
                WriteRowsEventData writeRowsEventData = event.getData();
                binlogEventProcessor.handleWrite(timestamp, masterCode, offsetMilliseconds, TABLE_MAP.get(writeRowsEventData.getTableId()), writeRowsEventData.getRows());
            } else if (EventType.isUpdate(eventType)) {
                UpdateRowsEventData updateRowsEventData = event.getData();
                binlogEventProcessor.handleUpdate(timestamp, masterCode, offsetMilliseconds, TABLE_MAP.get(updateRowsEventData.getTableId()), updateRowsEventData.getRows());
            } else {
                DeleteRowsEventData deleteRowsEventData = event.getData();
                binlogEventProcessor.handleDelete(timestamp, masterCode, offsetMilliseconds, TABLE_MAP.get(deleteRowsEventData.getTableId()), deleteRowsEventData.getRows());
            }
        } else if (EventType.ROTATE == eventType) {
            RotateEventData rotateEventData = event.getData();
            String binlogFilename = rotateEventData.getBinlogFilename();
            long binlogPosition = rotateEventData.getBinlogPosition();
            log.info("ROTATE File:{} Position:{}", binlogFilename, binlogPosition);
            binlogEventProcessor.rotateBinlogFile(masterCode, binlogFilename, binlogPosition);
        } else if (EventType.HEARTBEAT == eventType) {
            binlogEventProcessor.heartbeatTime(masterCode, new Date());
        }
    }
}